package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/11/26
 */
public class Demo10_TimeOut2
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                    .<WaterSensor>forMonotonousTimestamps()
                    .withTimestampAssigner((e, r) -> e.getTs());


                SingleOutputStreamOperator<WaterSensor> ds = env.readTextFile("data/cep2.txt")
                                                                .map(new MapFunction<String, WaterSensor>()
                                                               {
                                                                   @Override
                                                                   public WaterSensor map(String value) throws Exception {
                                                                       String[] words = value.split(",");
                                                                       return new WaterSensor(words[0], Long.valueOf(words[1]), Integer.valueOf(words[2]));
                                                                   }
                                                               })
                                                                .assignTimestampsAndWatermarks(watermarkStrategy);


                /*
                         .within(Time.seconds(2))
                         代表当前数据 ， 必须以规则1开始，下一条数据的时效范围是 seconds(2)。
                            当前数据必须是s1，下一条必须在2秒内达到，如果下一条超过2秒到达，就是当前数据超时了！

                            当前数据有以下情况
                                    ①未超时
                                            当前数据的下一条在 2s 内达到
                                                继续用next(s2)进行匹配。
                                                    匹配成功：  得到
                                                    匹配不成功:  得不到

                                    ②超时
                                            当前数据的下一条在 2s 内没达到。
                                                    可以得到
                 */
                Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("规则1")
                    .where(new SimpleCondition<WaterSensor>()
                    {
                        //定义匹配规则
                        @Override
                        public boolean filter(WaterSensor value) throws Exception {
                            return "s1".equals(value.getId());
                        }
                    })
                    .next("s2")
                    .where(new SimpleCondition<WaterSensor>()
                    {
                        //定义匹配规则
                        @Override
                        public boolean filter(WaterSensor value) throws Exception {
                            return "s2".equals(value.getId());
                        }
                    })
                    .within(Time.seconds(2));


                PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);

                //侧流
        OutputTag<String> outputTag = new OutputTag<>("late", TypeInformation.of(String.class));

        SingleOutputStreamOperator<String> ds2 = patternStream.process(new MyProcess());

        //匹配成功的
        ds2.print().setParallelism(1);
        //超时的，使用侧流
        ds2.getSideOutput(outputTag).printToErr().setParallelism(1);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    private static class MyProcess  extends PatternProcessFunction<WaterSensor, String> implements TimedOutPartialMatchHandler<WaterSensor>{

        //处理正常匹配
        @Override
        public void processMatch(Map<String, List<WaterSensor>> match, Context ctx, Collector<String> out) throws Exception {
                //输出到主流
                out.collect(match.toString());
        }

        //处理超时
        @Override
        public void processTimedOutMatch(Map<String, List<WaterSensor>> match, Context ctx) throws Exception {
            //输出到侧流
            ctx.output(new OutputTag<>("late", TypeInformation.of(String.class)),match.toString());
        }
    }
}
